package syM

import (
	"fmt"
	"db2s/Meta"
	"db2s/global"
	"db2s/go-log/log"
	"db2s/parDef"
	"db2s/ref"
	mq "db2s/topic-mq"
	"strings"
	"sync"
	"sync/atomic"
)

type GetObjectMeta struct {
	AlterStatus bool //判断是否为create,true为alter，false为create
	//JobResult       SrcDstJobObjectResult
	//ViewResult      SrcDstViewObjectResult
	SeqResult any
	//TriggerResult   SrcDstTriggerObjectResult
	//ForeignResult   SrcDstForeignObjectResult
	//FuncResult      SrcDstFuncObjectResult
	//ProcedureResult SrcDstProcedureObjectResult
	//TypeResult      SrcDstTypeObjectResult
	//PackageResult   SrcDstPackageObjectResult
	ObjectFixSql []fixSqlResult
}

func dropObject(plan *SchedulePlan2) (res any, err error) {
	var event = "[dropObject]"
	c := Meta.NewMeta([]string{GlobalPConfigs.dSns.SrcDBName, GlobalPConfigs.dSns.DestDBName}, "right", plan.Object.Role)
	c.Parameter1 = append(c.Parameter1, parDef.Parameter{Con: GlobalPConfigs.db.Source.GetDB("single"), Object: parDef.Object{Schema: plan.Object.Schema, Table: plan.Object.Table}})
	c.Parameter1 = append(c.Parameter1, parDef.Parameter{Con: GlobalPConfigs.db.Target.GetDB("single"), Object: parDef.Object{Schema: plan.Object.MSchema, Table: plan.Object.MTable}})
	log.MainLog().Info(fmt.Sprintf("(%v) %v Enable the \"drop object\" parameter and start deleting the target object %v", plan.Object.ObjectLogSeq, event, getTableName(plan.Object.Schema, plan.Object.Table)))
	if err = c.DDLObjectDrop(); err != nil {
		plan.errorAction(plan.Object.ObjectLogSeq, err)
		PlanContext.TPodTableStatusSwap(plan.TPod.Status, "error")
		log.ErrorLog().Error(fmt.Sprintf("(%v) %v Object %v failed to be deleted at the target end!!! error Msg: %v ", plan.Object.ObjectLogSeq, event, plan.Object.ObjectLogSeq, err))
		return
	}
	log.MainLog().Info(fmt.Sprintf("(%v) %v Object %v was successfully deleted at the target end.", plan.Object.ObjectLogSeq, event, getTableName(plan.Object.Schema, plan.Object.Table)))
	return
}

func objectBaseQueryListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var plan = message.Properties["plan"].(*SchedulePlan2)
	var wg sync.WaitGroup
	defer deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": ObjectBaseQuery, "outFay": GlobalPConfigs.result.Teletypewriter})
	defer func() {
		baseMsgSendBody.MsgCurryDone()
		baseMsgSendBody.MsgCurryWgDone()
	}()
	if metaGlobalInitActive(message) != nil {
		return mq.ReconsumeLater
	}
	plan.MQ.Wg = &wg
	atomic.SwapInt64(plan.MQ.FirstMsg.QuitExec, 1)
	plan.MQ.AbnormalQuit, plan.MQ.NormalQuit = global.EFunc{FuncName: mqStatusMonitorAbnormalQuit, Params: plan}, global.EFunc{FuncName: mqStatusMonitorNormalQuit, Params: plan}
	plan.MQ.MqTableStatusMonitor(getTableName(plan.Object.Schema, plan.Object.Table))
	if objectBaseQuery(message) != nil {
		return mq.ReconsumeLater
	}
	wg.Wait()
	return mq.ConsumeSuccess
}
func objectOptionsCheck(message mq.Message, options string) (result bool) {
	var plan = message.Properties["plan"].(*SchedulePlan2)
	return strings.EqualFold(plan.Object.Role, options) && func() bool {
		for _, v := range GlobalPConfigs.rules.Object.Object {
			if strings.EqualFold(v, options) {
				return true
			}
		}
		return false
	}()
}
func objectJobMetaData(message mq.Message) (res global.SrcDstJobObjectResult, err error) {
	//var objectMeta = message.Properties["objectMeta"].(*global.GetObjectMeta)
	var plan = message.Properties["plan"].(*SchedulePlan2)
	//if res, err = plan.getJobDefinitionBody(objectMeta.AlterStatus); err != nil {
	//	plan.errorAction(plan.Object.ObjectLogSeq, err)
	//	PlanContext.TPodTableStatusSwap(plan.TPod.Status, "error")
	//}
	var result any
	c := Meta.NewMeta([]string{GlobalPConfigs.dSns.SrcDBName, GlobalPConfigs.dSns.DestDBName}, "join", plan.Object.Role)
	c.Parameter1 = append(c.Parameter1, parDef.Parameter{Con: GlobalPConfigs.db.Source.GetDB("single"), Object: parDef.Object{Schema: plan.Object.Schema, Table: plan.Object.Table}})
	c.Parameter1 = append(c.Parameter1, parDef.Parameter{Con: GlobalPConfigs.db.Target.GetDB("single"), Object: parDef.Object{Schema: plan.Object.MSchema, Table: plan.Object.MTable}})
	if result, err = c.GetSeqMetaST(); err != nil {
		return
	}
	fmt.Println("-----------seq:", result)
	return
}
func objectIsJobMetaListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		plan       = message.Properties["plan"].(*SchedulePlan2)
		objectMeta = message.Properties["objectMeta"].(*GetObjectMeta)
		//err        error
	)
	defer deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": ObjectIsJobMeta, "outFay": GlobalPConfigs.result.Teletypewriter})
	//if objectOptionsCheck(message, "job") {
	//	if objectMeta.JobResult, err = objectJobMetaData(message); err != nil {
	//		return mq.ReconsumeLater
	//	}
	//}
	if !sendMsg(getCheckMod(), ObjectIsViewMeta, getTableName(plan.Object.Schema, plan.Object.Table), getMsgSendData(plan, map[string]any{
		"logSeq": plan.Object.ObjectLogSeq, "objectMeta": objectMeta, "topic": ObjectIsViewMeta,
	})) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}
func objectViewMetaData(message mq.Message) (res global.SrcDstViewObjectResult, err error) {
	//var objectMeta = message.Properties["objectMeta"].(*global.GetObjectMeta)
	//var plan = message.Properties["plan"].(*SchedulePlan2)
	//if GlobalPConfigs.rules.Object.Drop && objectMeta.AlterStatus {
	//	if _, err = dropObject(plan); err != nil {
	//		return
	//	}
	//	objectMeta.AlterStatus = false
	//}
	//if res, err = plan.getViewDefinitionBody(objectMeta.AlterStatus); err != nil {
	//	plan.errorAction(plan.Object.ObjectLogSeq, err)
	//	PlanContext.TPodTableStatusSwap(plan.TPod.Status, "error")
	//}
	return
}
func objectIsViewMetaListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		plan       = message.Properties["plan"].(*SchedulePlan2)
		objectMeta = message.Properties["objectMeta"].(*GetObjectMeta)
		//err        error
	)
	defer deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": ObjectIsViewMeta, "outFay": GlobalPConfigs.result.Teletypewriter})
	//if objectOptionsCheck(message, "view") {
	//	if objectMeta.ViewResult, err = objectViewMetaData(message); err != nil {
	//		return mq.ReconsumeLater
	//	}
	//}
	if !sendMsg(getCheckMod(), ObjectIsProcedureMeta, getTableName(plan.Object.Schema, plan.Object.Table), getMsgSendData(plan, map[string]any{
		"logSeq": plan.Object.ObjectLogSeq, "objectMeta": objectMeta, "topic": ObjectIsProcedureMeta,
	})) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}
func objectProcedureMetaData(message mq.Message) (res global.SrcDstProcedureObjectResult, err error) {
	//var objectMeta = message.Properties["objectMeta"].(*global.GetObjectMeta)
	//var plan = message.Properties["plan"].(*SchedulePlan2)
	//if res, err = plan.getProcedureDefinitionBody(objectMeta.AlterStatus); err != nil {
	//	plan.errorAction(plan.Object.ObjectLogSeq, err)
	//	PlanContext.TPodTableStatusSwap(plan.TPod.Status, "error")
	//}
	return
}
func objectIsProcedureMetaListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		plan       = message.Properties["plan"].(*SchedulePlan2)
		objectMeta = message.Properties["objectMeta"].(*GetObjectMeta)
		//	err        error
	)
	//defer deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": ObjectIsProcedureMeta, "outFay": GlobalPConfigs.result.Teletypewriter})
	//if objectOptionsCheck(message, "procedure") {
	//	if objectMeta.ProcedureResult, err = objectProcedureMetaData(message); err != nil {
	//		return mq.ReconsumeLater
	//	}
	//}
	if !sendMsg(getCheckMod(), ObjectIsFuncMeta, getTableName(plan.Object.Schema, plan.Object.Table), getMsgSendData(plan, map[string]any{
		"logSeq": plan.Object.ObjectLogSeq, "objectMeta": objectMeta, "topic": ObjectIsFuncMeta,
	})) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}
func objectFuncMetaData(message mq.Message) (res global.SrcDstFuncObjectResult, err error) {
	var objectMeta = message.Properties["objectMeta"].(*GetObjectMeta)
	var plan = message.Properties["plan"].(*SchedulePlan2)
	if res, err = plan.getFuncDefinitionBody(objectMeta.AlterStatus); err != nil {
		plan.errorAction(plan.Object.ObjectLogSeq, err)
		PlanContext.TPodTableStatusSwap(plan.TPod.Status, "error")
	}
	return
}
func objectIsFuncMetaListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		plan       = message.Properties["plan"].(*SchedulePlan2)
		objectMeta = message.Properties["objectMeta"].(*GetObjectMeta)
		//err        error
	)
	defer deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": ObjectIsFuncMeta, "outFay": GlobalPConfigs.result.Teletypewriter})
	//if objectOptionsCheck(message, "func") {
	//	if objectMeta.FuncResult, err = objectFuncMetaData(message); err != nil {
	//		return mq.ReconsumeLater
	//	}
	//}
	if !sendMsg(getCheckMod(), ObjectIsTriggerMeta, getTableName(plan.Object.Schema, plan.Object.Table), getMsgSendData(plan, map[string]any{
		"logSeq": plan.Object.ObjectLogSeq, "objectMeta": objectMeta, "topic": ObjectIsTriggerMeta,
	})) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}
func objectTriggerMetaData(message mq.Message) (res global.SrcDstTriggerObjectResult, err error) {
	//var objectMeta = message.Properties["objectMeta"].(*global.GetObjectMeta)
	//var plan = message.Properties["plan"].(*SchedulePlan2)
	//if res, err = plan.getTriggerDefinitionBody(objectMeta.AlterStatus); err != nil {
	//	plan.errorAction(plan.Object.ObjectLogSeq, err)
	//	PlanContext.TPodTableStatusSwap(plan.TPod.Status, "error")
	//}
	return
}
func objectIsTriggerMetaListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		plan       = message.Properties["plan"].(*SchedulePlan2)
		objectMeta = message.Properties["objectMeta"].(*GetObjectMeta)
		//err        error
	)
	defer deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": ObjectIsTriggerMeta, "outFay": GlobalPConfigs.result.Teletypewriter})
	//if objectOptionsCheck(message, "trigger") {
	//	if objectMeta.TriggerResult, err = objectTriggerMetaData(message); err != nil {
	//		return mq.ReconsumeLater
	//	}
	//}
	if !sendMsg(getCheckMod(), ObjectIsPackageMeta, getTableName(plan.Object.Schema, plan.Object.Table), getMsgSendData(plan, map[string]any{
		"logSeq": plan.Object.ObjectLogSeq, "objectMeta": objectMeta, "topic": ObjectIsPackageMeta,
	})) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}
func objectIsPackageMetaData(message mq.Message) (res global.SrcDstPackageObjectResult, err error) {
	var objectMeta = message.Properties["objectMeta"].(*global.GetObjectMeta)
	var plan = message.Properties["plan"].(*SchedulePlan2)
	if res, err = plan.getPackageDefinitionBody(objectMeta.AlterStatus); err != nil {
		plan.errorAction(plan.Object.ObjectLogSeq, err)
		PlanContext.TPodTableStatusSwap(plan.TPod.Status, "error")
	}
	return
}
func objectIsPackageMetaListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		plan       = message.Properties["plan"].(*SchedulePlan2)
		objectMeta = message.Properties["objectMeta"].(*GetObjectMeta)
		//err        error
	)
	defer deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": ObjectIsPackageMeta, "outFay": GlobalPConfigs.result.Teletypewriter})
	//if objectOptionsCheck(message, "package") {
	//	if objectMeta.PackageResult, err = objectIsPackageMetaData(message); err != nil {
	//		return mq.ReconsumeLater
	//	}
	//}
	if !sendMsg(getCheckMod(), ObjectIsTypeMeta, getTableName(plan.Object.Schema, plan.Object.Table), getMsgSendData(plan, map[string]any{
		"logSeq": plan.Object.ObjectLogSeq, "objectMeta": objectMeta, "topic": ObjectIsTypeMeta,
	})) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}
func objectTypeMetaData(message mq.Message) (res global.SrcDstTypeObjectResult, err error) {
	var objectMeta = message.Properties["objectMeta"].(*global.GetObjectMeta)
	var plan = message.Properties["plan"].(*SchedulePlan2)
	if res, err = plan.getTypeDefinitionBody(objectMeta.AlterStatus); err != nil {
		plan.errorAction(plan.Object.ObjectLogSeq, err)
		PlanContext.TPodTableStatusSwap(plan.TPod.Status, "error")
	}
	return
}
func objectIsTypeMetaListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		plan       = message.Properties["plan"].(*SchedulePlan2)
		objectMeta = message.Properties["objectMeta"].(*GetObjectMeta)
		//err        error
	)
	defer deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": ObjectIsTypeMeta, "outFay": GlobalPConfigs.result.Teletypewriter})
	//if objectOptionsCheck(message, "type") {
	//	if objectMeta.TypeResult, err = objectTypeMetaData(message); err != nil {
	//		return mq.ReconsumeLater
	//	}
	//}
	if !sendMsg(getCheckMod(), ObjectIsForeignMeta, getTableName(plan.Object.Schema, plan.Object.Table), getMsgSendData(plan, map[string]any{
		"logSeq": plan.Object.ObjectLogSeq, "objectMeta": objectMeta, "topic": ObjectIsForeignMeta,
	})) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}
func objectForeignMetaData(message mq.Message) (res global.SrcDstForeignObjectResult, err error) {
	var objectMeta = message.Properties["objectMeta"].(*global.GetObjectMeta)
	var plan = message.Properties["plan"].(*SchedulePlan2)
	if res, err = plan.getForeignDefinitionBody(objectMeta.AlterStatus); err != nil {
		plan.errorAction(plan.Object.ObjectLogSeq, err)
		PlanContext.TPodTableStatusSwap(plan.TPod.Status, "error")
	}
	return
}
func objectIsForeignMetaListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		plan       = message.Properties["plan"].(*SchedulePlan2)
		objectMeta = message.Properties["objectMeta"].(*GetObjectMeta)
		//err        error
	)
	defer deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": ObjectIsForeignMeta, "outFay": GlobalPConfigs.result.Teletypewriter})
	//if objectOptionsCheck(message, "foreign") {
	//	if objectMeta.ForeignResult, err = objectForeignMetaData(message); err != nil {
	//		return mq.ReconsumeLater
	//	}
	//}
	if !sendMsg(getCheckMod(), ObjectIsSeqMeta, getTableName(plan.Object.Schema, plan.Object.Table), getMsgSendData(plan, map[string]any{
		"logSeq": plan.Object.ObjectLogSeq, "objectMeta": objectMeta, "topic": ObjectIsSeqMeta,
	})) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}

func objectIsSeqMetaListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		plan       = message.Properties["plan"].(*SchedulePlan2)
		objectMeta = message.Properties["objectMeta"].(*GetObjectMeta)
		err        error
	)
	defer deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": ObjectIsSeqMeta, "outFay": GlobalPConfigs.result.Teletypewriter})
	if objectOptionsCheck(message, "seq") {
		if objectMeta.SeqResult, err = objectSeqMetaData(message); err != nil {
			return mq.ReconsumeLater
		}
	}
	if !sendMsg(getCheckMod(), ObjectIsMetaComparison, getTableName(plan.Object.Schema, plan.Object.Table), getMsgSendData(plan, map[string]any{
		"logSeq": plan.Object.ObjectLogSeq, "objectMeta": objectMeta, "topic": ObjectIsMetaComparison,
	})) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}
func objectIsMetaComparisonListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		plan       = message.Properties["plan"].(*SchedulePlan2)
		objectMeta = message.Properties["objectMeta"].(*GetObjectMeta)
	)
	defer deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": ObjectIsMetaComparison, "outFay": GlobalPConfigs.result.Teletypewriter})
	if objectMeta.AlterStatus { //是校验状态不是create table状态
		//	if len(tableMeta.ColumnMeta) > 0 {
		//		tableMeta.MetaAbnormal = plan.TableMetaAbnormalComparison(tableMeta.ColumnMeta)
		//	}
		//	if len(tableMeta.IndexMeta) > 0 {
		//		tableMeta.IndexAbnormal = plan.TableIndexAbnormalComparison(tableMeta.IndexMeta)
		//	}
		//	if len(tableMeta.PartitionMeta) > 0 {
		//		tableMeta.PartitionAbnormal = plan.TablePartitionAbnormalComparison(tableMeta.PartitionMeta)
		//	}
	}
	if !sendMsg(getCheckMod(), ObjectIsFix, getTableName(plan.Object.Schema, plan.Object.Table), getMsgSendData(plan, map[string]any{
		"logSeq": plan.Object.ObjectLogSeq, "objectMeta": objectMeta, "topic": ObjectIsFix,
	})) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}

func objectIsFixListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var (
		plan       = message.Properties["plan"].(*SchedulePlan2)
		objectMeta = message.Properties["objectMeta"].(*GetObjectMeta)
		err        error
	)
	defer deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": ObjectIsFix, "outFay": GlobalPConfigs.result.Teletypewriter})
	if objectMeta.AlterStatus { //是校验状态不是create table状态
		//	plan.TPod.Object.ExecObject = "alter"
		//	if result, err = plan.alterFixSql(tableMeta, plan.Object.ObjectLogSeq); err != nil {
		//		plan.errorAction(message, plan.Object.ObjectLogSeq, err)
		//		return mq.ReconsumeLater
		//	}
	} else {
		plan.TPod.Object.ExecObject = "create"
		if objectMeta.ObjectFixSql, err = plan.objectCreateFixSql(plan, objectMeta); err != nil {
			return mq.ReconsumeLater
		}
	}
	if !sendMsg(getCheckMod(), ObjectIsApply, getTableName(plan.Object.Schema, plan.Object.Table), getMsgSendData(plan, map[string]any{
		"logSeq": plan.Object.ObjectLogSeq, "fixMeta": message.Properties["tableMeta1"], "topic": ObjectIsApply,
		"fixSqlResult": objectMeta.ObjectFixSql,
	})) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}
func objectIsApplyListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var event = "[objectIsApplyListeners]"
	defer func() { _ = ref.RecoverPanic(event, recover()) }()
	var plan = message.Properties["plan"].(*SchedulePlan2)
	defer func() {
		deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": ObjectIsApply, "outFay": GlobalPConfigs.result.Teletypewriter})
		plan.MQ.FirstMsgCustomerAdd()
		plan.MQ.FirstObjectCustomerAdd()
		plan.MQ.SecondMsgCustomerAdd()
	}()
	if err := execFixSql(message); err != nil {
		plan.errorAction(plan.Object.ObjectLogSeq, err)
		PlanContext.TPodTableStatusSwap(plan.TPod.Status, "abnormal")
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}
